package org.weishe.pagerank;

import java.io.IOException;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.StringUtils;

public class PRReducer extends Reducer<Text, Text, NullWritable, Text> {
	private double nWeight;// 新权重
	private double q = 0.85;// 阻力系数
	private Text ovalue = new Text();
	private String people = "";
	private int count = 0;

	protected void setup(Context context) throws IOException, InterruptedException {
		count = Integer.parseInt(context.getConfiguration().get("sumcount"));
	}

	@Override
	protected void reduce(Text key, Iterable<Text> value, Reducer<Text, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		double wsum = 0;
		for (Text v : value) {
			String vs = v.toString();
			if (!vs.startsWith(",")) {

				String[] item = StringUtils.split(vs, ',');
				double w = Double.parseDouble(item[1]);
				wsum = wsum + w;
			} else {
				people = vs;
			}
		}
		nWeight = wsum * q + (1 - q) / count;
		ovalue.set(key.toString() + "," + nWeight + people);
		context.write(NullWritable.get(), ovalue);
	}

}
